package com.psychologyCommunity.Community.event;

import com.alibaba.fastjson.JSONObject;
import com.psychologyCommunity.Community.entity.DiscussPost;
import com.psychologyCommunity.Community.entity.Event;
import com.psychologyCommunity.Community.entity.Message;
import com.psychologyCommunity.Community.service.DiscussPostService;
import com.psychologyCommunity.Community.service.ElasticsearchService;
import com.psychologyCommunity.Community.service.MessageService;
import com.psychologyCommunity.Community.util.CommunityConstance;
import org.apache.kafka.clients.consumer.ConsumerRecord;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

@Component
public class EventConsumer implements CommunityConstance {
    //记日志 ， 处理事件，发布消息这个过程 可能会有一些隐含的问题出现
    private static final Logger logger=LoggerFactory.getLogger(EventConsumer.class);
    //处理事件最终是要给 某个人发送一条消息，发消息 最终是要往message表里面插入一条数据
    @Autowired
    private MessageService messageService;
    //可以一个方法消费多个主题，一个主题也可以被多个方法消费
    //此时三个主题都被消费
    @KafkaListener(topics ={TOPIC_COMMENT,TOPIC_LIKE,TOPIC_FOLLOW})
    public void handleCommentMessage(ConsumerRecord record) {
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空！");
            return;
        }
        //解析字符串 转换格式
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {   //转完之后 如果为空 即虽然有值，但是还原不回来，那就是格式不对
            logger.error("消息格式错误");
            return;
        }

        //发送站内通知--主要就是要构造一个Json对象
        Message message=new Message();
        message.setFromId(SYSTEM_USER_ID);
        message.setToId(event.getEntityUserId());
        message.setConversationId(event.getTopic());
        message.setCreateTime(new Date());
        // 事件的消费者在消费这个事件的时候得到最原始的数据，最终需要把这个数据转换成一个message
        //messsage中包含了一些基础数据 上面的 ，以及内容content，content最终是要拼出页面上的预浏览的句子
        Map<String,Object> content=new HashMap<>();
        content.put("userId",event.getUserId());//事件是谁触发的
        content.put("entityType",event.getEntityType());//是对哪个实体--帖子，评价 以及回复。
        content.put("entityId",event.getEntityId());
        if(!event.getData().isEmpty()){
            for (Map.Entry<String, Object> entry:event.getData().entrySet()){
                content.put(entry.getKey(),entry.getValue());
            }
        }
        message.setContent(JSONObject.toJSONString(content));
        messageService.addMessage(message);

    }
    //消费删帖事件
//    @KafkaListener(topics ={TOPIC_DELETE})
//    public void handleDeleteMessage(ConsumerRecord record) {
//        if (record == null || record.value() == null) {
//            logger.error("消息的内容为空！");
//            return;
//        }
//        //解析字符串 转换格式
//        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
//        if (event == null) {   //转完之后 如果为空 即虽然有值，但是还原不回来，那就是格式不对
//            logger.error("消息格式错误");
//            return;
//        }
//    }

    @Value("${wk.image.command}")
    private  String wkImageCommand;

    @Value("${wk.image.storage}")
    private  String wkImageStorage;

//    消费 分享事件
    @KafkaListener(topics ={TOPIC_SHARE})
    public void handleShareMessage(ConsumerRecord record) {
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空！");
            return;
        }
        //解析字符串 转换格式
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {   //转完之后 如果为空 即虽然有值，但是还原不回来，那就是格式不对
            logger.error("消息格式错误");
            return;
        }
        //取到这三个值 ，然后去拼 命令cmd
        String htmlUrl= (String) event.getData().get("htmlUrl");
        String fileName= (String) event.getData().get("fileName");
        String suffix= (String) event.getData().get("suffix");

        //" --quality 75 " 表示前后的压缩比，质量，不要忘记空格
        String cmd=wkImageCommand+" --quality 75 "+htmlUrl+" "
                +wkImageStorage+"/"+fileName+suffix;
        try {
            Runtime.getRuntime().exec(cmd);//执行这个命令
            logger.info("宝儿，生成长图成功："+ cmd);
        } catch (IOException e) {
            e.printStackTrace();
            logger.error("宝儿，生成长图失败："+ e.getMessage());
        }

    }
    @Autowired
    private DiscussPostService discussPostService;
    @Autowired
    private ElasticsearchService elasticsearchService;
    //消费发帖事件  --只消费topic=publish的
    @KafkaListener(topics = {TOPIC_PUBLISH})
    public void handlePublishMessage(ConsumerRecord record){
        if (record == null || record.value() == null) {
            logger.error("消息的内容为空！");
            return;
        }
        //解析字符串 转换格式
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (event == null) {   //转完之后 如果为空 即虽然有值，但是还原不回来，那就是格式不对
            logger.error("消息格式错误");
            return;
        }
        DiscussPost discussPostById = discussPostService.findDiscussPostById(event.getEntityId());//从事件中得到帖子的id 然后利用这个id去查
        elasticsearchService.saveDiscussPost(discussPostById);//存到ES中
    }



}
